package com.study.reactor.create;

import io.reactivex.rxjava3.annotations.NonNull;
import io.reactivex.rxjava3.core.Observable;
import io.reactivex.rxjava3.core.ObservableEmitter;
import io.reactivex.rxjava3.core.ObservableOnSubscribe;
import io.reactivex.rxjava3.core.Observer;
import io.reactivex.rxjava3.disposables.Disposable;
import io.reactivex.rxjava3.functions.Action;
import io.reactivex.rxjava3.functions.Consumer;

public class CreateEventImpl implements CreateEvent{

    public static void main(String[] args) {
        consumer();

    }

    public static void consumer(){
        Observable.create(new ObservableOnSubscribe<String>() {
            @Override
            public void subscribe(@NonNull ObservableEmitter<String> emitter) throws Throwable {

                emitter.onNext("111");
                emitter.onNext("222");

                emitter.onError(new Throwable("aaaa"));
                System.out.println("aaa");
                System.out.println("comm");
                emitter.onNext("333");
                System.out.println("333");
                emitter.onComplete();
            }
        }).subscribe(new Consumer<String>() {
            @Override
            public void accept(String s) throws Throwable {
                System.out.println("处理"+s);
            }
        }, new Consumer<Throwable>() {
            @Override
            public void accept(Throwable throwable) throws Throwable {
                System.out.println("处理"+throwable.getMessage());
            }
        }, new Action() {
            @Override
            public void run() throws Throwable {
                System.out.println("完成");
            }
        });
    }


    public static void Observer(){
        Observable.create(new ObservableOnSubscribe<String>() {
            @Override
            public void subscribe(@NonNull ObservableEmitter<String> emitter) throws Throwable {

                emitter.onNext("111");
                emitter.onNext("222");
                emitter.onError(new Throwable("aaaa"));
                System.out.println("aaa");
                emitter.onComplete();
                System.out.println("comm");
                emitter.onNext("333");
                System.out.println("333");
            }
        }).subscribe(new Observer<String>() {
            @Override
            public void onSubscribe(@NonNull Disposable d) {
                System.out.println("终止"+d.isDisposed());
            }

            @Override
            public void onNext(@NonNull String s) {
                System.out.println("处理"+s);
            }

            @Override
            public void onError(@NonNull Throwable e) {
                System.out.println("出错"+e.getMessage());
            }

            @Override
            public void onComplete() {
                System.out.println("完成");
            }
        });
    }
}
